

package org.example.datastream.connector;

import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.example.datastream.test.custom.User;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;

/**
 * JDBC connector be sink only
 */
public class Jdbc {

	public static void main(String[] args) throws Exception {
		// set up the streaming execution environment
		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		Integer limit = 10;
		if (null != args && args.length > 0) {
			limit = Integer.valueOf(args[0]);
		}

		List<User> list = new ArrayList<>(10);
		for (int i = 1; i < limit; i++) {
			String username = "user-" + i;
			String password = "passwd-" + i;
			String nickname = "nick-" + i;
			String sex = i%2==1?"男":"女";
			int age = 20 + i % 10;
			User user = new User(i, username, password, nickname, sex, age);
			list.add(user);
		}

		env.fromCollection(list)
				.addSink(JdbcSink.sink(
				"insert into t_users(id,username,password,nickname,sex,age) values(?,?,?,?,?,?)",
					new JdbcStatementBuilder<User>() {
						@Override
						public void accept(PreparedStatement ps, User user) throws SQLException {
							ps.setInt(1, user.getId());
							ps.setString(2, user.getUsername());
							ps.setString(3, user.getPassword());
							ps.setString(4, user.getNickname());
							ps.setString(5, user.getSex());
							ps.setInt(6, user.getAge());
						}
					},
					new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
							.withUrl("jdbc:mysql://localhost:3306/db_test?useSSL=false&useUnicode=true&characterEncoding=UTF-8")
							.withDriverName("com.mysql.jdbc.Driver")
							.withUsername("flinksql")
							.withPassword("w0c53w0c53")
							.build()
				))
				.name("t_users")
				.setParallelism(2);

		// execute program
		env.execute("Flink Streaming Java API Skeleton");
	}
}
